package gbench.sandbox.nsql.mongo;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

import org.bson.Document;
import org.bson.conversions.Bson;
import org.bson.types.ObjectId;

import com.mongodb.BasicDBObject;
import com.mongodb.MongoClient;
import com.mongodb.MongoClientOptions;
import com.mongodb.ServerAddress;
import com.mongodb.MongoClientOptions.Builder;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoCursor;
import com.mongodb.client.MongoDatabase;
import com.mongodb.client.MongoIterable;
import com.mongodb.client.model.Filters;
import com.mongodb.client.result.DeleteResult;

import gbench.common.matlib.algebra.lisp.DFrame;
import gbench.common.matlib.algebra.lisp.IRecord;

/**
 * 
 * @author Administrator
 *
 */
public class MongoDB {

    /**
     * MongoDB 构造函数
     * 
     * @param host 主机地址
     * @param port 端口
     */
    public MongoDB(final String host, final int port) {

        // 大部分用户使用mongodb都在安全内网下，
        // 但如果将mongodb设为安全验证模式，就需要在客户端提供用户名和密码：
        // boolean auth = db.authenticate(myUserName, myPassword);
        Builder options = new MongoClientOptions.Builder();
        options.cursorFinalizerEnabled(true);
        options.connectionsPerHost(300);// 连接池设置为300个连接,默认为100
        options.connectTimeout(30000);// 连接超时，推荐>3000毫秒
        options.maxWaitTime(5000); //
        options.socketTimeout(0);// 套接字超时时间，0无限制
        mongoClient = new MongoClient(new ServerAddress(host, port), options.build());
    }

    // ------------------------------------共用方法---------------------------------------------------
    /**
     * 获取DB实例 - 指定DB
     *
     * @param dbName
     * @return
     */
    public MongoDatabase getDB(final String dbName) {
        if (dbName != null && !"".equals(dbName)) {
            MongoDatabase database = mongoClient.getDatabase(dbName);
            return database;
        }
        return null;
    }

    /**
     * 
     * @param line
     * @return
     */
    public String[] line2clcname(final String line) {
        final var ss = line.split("[\\.]");
        return ss.length > 1 ? ss : new String[] { "test", line };
    }

    /**
     * 获取collection对象 - 指定Collection
     *
     * @param line 集合名
     * @return 数据集合
     */
    public MongoCollection<Document> getCollection(final String line) {
        final var ss = this.line2clcname(line);
        return this.getCollection(ss[0], ss[1]);
    }

    /**
     * 获取collection对象 - 指定Collection
     *
     * @param dbName   数据库名
     * @param collName 集合名
     * @return
     */
    public MongoCollection<Document> getCollection(final String dbName, final String collName) {
        if (null == collName || "".equals(collName)) {
            return null;
        }
        if (null == dbName || "".equals(dbName)) {
            return null;
        }
        MongoCollection<Document> collection = mongoClient.getDatabase(dbName).getCollection(collName);
        return collection;
    }

    /**
     * 查询DB下的所有表名
     */
    public List<String> getAllCollections(final String dbName) {
        MongoIterable<String> colls = getDB(dbName).listCollectionNames();
        List<String> _list = new ArrayList<String>();
        for (String s : colls) {
            _list.add(s);
        }
        return _list;
    }

    /**
     * 获取所有数据库名称列表
     *
     * @return
     */
    public MongoIterable<String> getAllDBNames() {
        MongoIterable<String> s = mongoClient.listDatabaseNames();
        return s;
    }

    /**
     * 删除一个数据库
     */
    public void dropDB(final String dbName) {
        getDB(dbName).drop();
    }

    /**
     * 查找对象 - 根据主键_id
     *
     * @param collection
     * @param id
     * @return
     */
    public Document findById(final MongoCollection<Document> coll, final String id) {
        ObjectId _idobj = null;
        try {
            _idobj = new ObjectId(id);
        } catch (Exception e) {
            return null;
        }
        Document myDoc = coll.find(Filters.eq("_id", _idobj)).first();
        return myDoc;
    }

    /** 统计数 */
    public int getCount(final MongoCollection<Document> coll) {
        int count = (int) coll.countDocuments();
        return count;
    }

    /** 条件查询 */
    public MongoCursor<Document> find(final MongoCollection<Document> coll, final Bson filter) {
        return coll.find(filter).iterator();
    }

    /** 分页查询 */
    public MongoCursor<Document> findByPage(final MongoCollection<Document> coll, final Bson filter, final int pageNo,
            final int pageSize) {
        Bson orderBy = new BasicDBObject("_id", 1);
        return coll.find(filter).sort(orderBy).skip((pageNo - 1) * pageSize).limit(pageSize).iterator();
    }

    /**
     * 通过ID删除
     *
     * @param coll
     * @param id
     * @return
     */
    public int deleteById(final MongoCollection<Document> coll, final String id) {
        int count = 0;
        ObjectId _id = null;
        try {
            _id = new ObjectId(id);
        } catch (Exception e) {
            return 0;
        }
        Bson filter = Filters.eq("_id", _id);
        DeleteResult deleteResult = coll.deleteOne(filter);
        count = (int) deleteResult.getDeletedCount();
        return count;
    }

    /**
     * 数据更新
     *
     * @param coll   集合对象
     * @param id     数据ID
     * @param newdoc 新文档
     * @return 增量更新数据
     */
    public Document updateById(final MongoCollection<Document> coll, final String id, final Document newdoc) {
        ObjectId _idobj = null;
        try {
            _idobj = new ObjectId(id);
        } catch (Exception e) {
            return null;
        }
        Bson filter = Filters.eq("_id", _idobj);
        // coll.replaceOne(filter, newdoc); // 完全替代
        coll.updateOne(filter, new Document("$set", newdoc));
        return newdoc;
    }

    /**
     * 集合流
     * 
     * @param line    数据名
     * @param maxsize 最大数值
     * @return 数据流
     */
    public Stream<IRecord> recordS(final String line) {
        final var ss = this.line2clcname(line);
        final String dbName = ss[0];
        final String collName = ss[1];
        return this.recordS(dbName, collName, null);
    }

    /**
     * 集合流
     * 
     * @param line    数据名
     * @param maxsize 最大数值
     * @return 数据流
     */
    public Stream<IRecord> recordS(final String line, final Integer maxsize) {
        final var ss = this.line2clcname(line);
        final String dbName = ss[0];
        final String collName = ss[1];
        return this.recordS(dbName, collName, maxsize);
    }

    /**
     * 集合流
     * 
     * @param dbName   数据名
     * @param collName 集合名
     * @param maxsize  最大数值
     * @return 数据流
     */
    public Stream<IRecord> recordS(final String dbName, final String collName, final Integer maxsize) {
        final var collection = this.getCollection(dbName, collName);
        final var spliterator = maxsize == null //
                ? collection.find().spliterator() // final all
                : collection.find().limit(maxsize).spliterator();
        final var stream = StreamSupport.stream(spliterator, false).map(IRecord::REC);
        return stream;
    }

    /**
     * 集合流
     * 
     * @param line 集合名
     * @return 数据框
     */
    public DFrame dframe(final String line) {
        final var ss = this.line2clcname(line);
        final String dbName = ss[0];
        final String collName = ss[1];
        return this.recordS(dbName, collName, null).collect(DFrame.dfmclc);
    }

    /**
     * 集合流
     * 
     * @param dbName   数据名
     * @param collName 集合名
     * @return 数据框
     */
    public DFrame dframe(final String dbName, final String collName) {
        return this.recordS(dbName, collName, null).collect(DFrame.dfmclc);
    }

    /**
     * 集合流
     * 
     * @param dbName   数据名
     * @param collName 集合名
     * @return 数据框
     */
    public DFrame dframe(final String dbName, final String collName, final Integer maxsize) {
        return this.recordS(dbName, collName, null).collect(DFrame.dfmclc);
    }

    /**
     * 
     * @param dbName   数据库
     * @param collName
     */
    public void dropCollection(final String dbName, final String collName) {
        getDB(dbName).getCollection(collName).drop();
    }

    /**
     * 
     * @param dbName   数据库
     * @param collName
     */
    public void dropCollection(final String line) {
        final var ss = this.line2clcname(line);
        final String dbName = ss[0];
        final String collName = ss[1];
        this.dropCollection(dbName, collName);
    }

    /**
     * 关闭Mongodb
     */
    public void close() {
        if (mongoClient != null) {
            mongoClient.close();
            mongoClient = null;
        }
    }

    private MongoClient mongoClient; // 客户端
}
